package com.swhy

import com.utils._
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import org.codehaus.jackson.map.ObjectMapper

import java.net.URI
import java.util.{Date, UUID}


object text_scala {
  def main(args: Array[String]): Unit = {
//    if (args.length != 3) {
//      println(
//        """
//          |com.swhy.CompanyInfo
//          |params:
//          | InputDate    刷数日期
//          | OutputPath   输出路径
//          | SplitNum     文件生成数量
//        """.stripMargin)
//      sys.exit()
//    }
    val conf = new Configuration()
 val Array(inputDt, outputPath, splitNum) = args
    val sparkConf = new SparkConf().setAppName(this.getClass.getName)
    val sc = new SparkContext(sparkConf)
    val hiveContext = new HiveContext(sc)
//改下方的sql为需要装换成jason的sql
    val allcompany =s"""
        |select
        |asset_acct,
        |biz_dt,
        |cust_cd,
        |intg_cust_id,
        |belong_org_id,
        |deve_org_id,
        |crdt_acct_flag,
        |cap_bal,
        |in_tran_amt,
        |biz_date
        |from sum_data.sum_t98_asset_acct_info_z
      """.stripMargin
    print(allcompany)
    val frame = hiveContext.sql(allcompany)
    val result = frame.rdd.map(x => {
      var cominfo = new StringBuilder()

      val asset_acct = x.getAs[String]("asset_acct")
      val biz_dt = x.getAs[String]("biz_dt")
      val cust_cd = x.getAs[String]("cust_cd")
      val intg_cust_id = x.getAs[String]("intg_cust_id")
      val belong_org_id = x.getAs[String]("belong_org_id")
      val deve_org_id = x.getAs[String]("deve_org_id")
      val crdt_acct_flag = x.getAs[String]("crdt_acct_flag")

      val cap_bal =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("cap_bal"))
      val cap_bal1 = cap_bal.bigDecimal
      val in_tran_amt =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("in_tran_amt"))
      val in_tran_amt1 = in_tran_amt.bigDecimal

      val biz_date = x.getAs[Integer]("biz_date")

      val _id = new ID()
      val index = new Index()
      val companyInfo = new text()

      _id.set_id(UUID.randomUUID().toString)
      index.setIndex(_id)
      companyInfo.setAssetAcct(asset_acct)
      companyInfo.setCustCd(cust_cd)
      companyInfo.setIntgCustId(intg_cust_id)
      companyInfo.setBelongOrgId(belong_org_id)
      companyInfo.setDeveOrgId(deve_org_id)
      companyInfo.setCrdtAcctFlag(crdt_acct_flag)
      companyInfo.setCapBal(cap_bal1)
      companyInfo.setInTranSmt(in_tran_amt1)
      companyInfo.setBizDate(biz_date)
      companyInfo.setBizDt(biz_dt)


      val mapper = new ObjectMapper()

      //val indexJson = mapper.writeValueAsString(index) //企业自增编码json串
      val companyInfoJson = mapper.writeValueAsString(companyInfo) //企业信息串
      //cominfo = cominfo.append(indexJson + "\n" + companyInfoJson)
      var list =  new java.util.ArrayList[String]();
      list.add(companyInfoJson);
      val es_ts_cs =new es_ts()
      es_ts_cs.addDataBatch(list,"test_data_index")

    })

    sc.stop()
  }
}
